package cn.yto.rocketmq.rocketmqselect.task;


import cn.yto.rocketmq.rocketmqselect.annotation.MultiMQAdminCmdMethod;
import cn.yto.rocketmq.rocketmqselect.config.RMQConfigure;
import cn.yto.rocketmq.rocketmqselect.service.CollectService;
import cn.yto.rocketmq.rocketmqselect.util.JsonUtil;
import com.google.common.base.Stopwatch;
import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.protocol.body.BrokerStatsData;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
import org.apache.rocketmq.common.protocol.body.GroupList;
import org.apache.rocketmq.common.protocol.body.TopicList;
import org.apache.rocketmq.common.protocol.route.BrokerData;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
import org.apache.rocketmq.tools.admin.MQAdminExt;
import org.apache.rocketmq.tools.command.stats.StatsAllSubCommand;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

@Component
public class CollectTask {

    @Resource
    private MQAdminExt mqAdminExt;
    @Resource
    private RMQConfigure rmqConfigure;

    @Resource
    private CollectService collectService;

    private final static Logger log = LoggerFactory.getLogger(CollectTask.class);

    public static String brokerVIPChannel(final String brokerAddr) {
            String[] ipAndPort = brokerAddr.split(":");
            String brokerAddrNew = ipAndPort[0] + ":" + (Integer.parseInt(ipAndPort[1]) - 2);
            return brokerAddrNew;

    }


   // @Scheduled(cron = "*/30 * * * * ?")
    @MultiMQAdminCmdMethod(timeoutMillis = 5000)
    public void collectTopic() {
        if (!rmqConfigure.isEnableDashBoardCollect()) {
            return;
        }

        Date date = new Date();
        Stopwatch stopwatch = Stopwatch.createUnstarted();
        try {
            TopicList topicList = mqAdminExt.fetchAllTopicList();
          /*  ClusterInfo clusterInfo = mqAdminExt.examineBrokerClusterInfo();
            HashMap<String, BrokerData> brokerAddrTable = clusterInfo.getBrokerAddrTable();
            List<BrokerData> ll = brokerAddrTable.values().stream().collect(Collectors.toList());
            List<String> valus = new ArrayList<>();
            List<HashMap<Long, String>> collect = ll.stream().map(BrokerData::getBrokerAddrs).collect(Collectors.toList());
            if (collect.size() > 0) {
                for (HashMap<Long, String> hm : collect) {//遍历外层List得到HashMap
                    Set<Long> set = hm.keySet();//得到存在Hashmap里面的key，即手机号
                    for (Long key : set) {
                        String s = hm.get(key);
                        valus.add(s);
                    }
                }
            }
            List<String> collect1 = valus.stream().map(s -> brokerVIPChannel(s)).collect(Collectors.toList());*/


            Set<String> topicSet = topicList.getTopicList();

            for (String topic : topicSet) {
                if (topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)
                        || topic.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX)
                        ) {
                    continue;
                }
                TopicRouteData topicRouteData = mqAdminExt.examineTopicRouteInfo(topic);


                GroupList groupList = mqAdminExt.queryTopicConsumeByWho(topic);

                double inTPS = 0;

                long inMsgCntToday = 0;

                double outTPS = 0;

                long outMsgCntToday = 0;

                for (BrokerData bd : topicRouteData.getBrokerDatas()) {
                    String masterAddr = bd.getBrokerAddrs().get(MixAll.MASTER_ID);
                    if (masterAddr != null) {
                        try {
                            stopwatch.start();
                            log.info("start time: {}", stopwatch.toString());
                            BrokerStatsData bsd = mqAdminExt.viewBrokerStatsData(masterAddr, BrokerStatsManager.TOPIC_PUT_NUMS, topic);
                            stopwatch.stop();
                            log.info("stop time : {}", stopwatch.toString());

                            inTPS += bsd.getStatsMinute().getTps();
                            inMsgCntToday += StatsAllSubCommand.compute24HourSum(bsd);
                        }
                        catch (Exception e) {
                            stopwatch.reset();
                            log.warn("Exception caught: mqAdminExt get broker stats data TOPIC_PUT_NUMS failed");
                            log.warn("Response [{}] ", e.getMessage());
                        }
                    }
                }

                if (groupList != null && !groupList.getGroupList().isEmpty()) {

                    for (String group : groupList.getGroupList()) {
                        for (BrokerData bd : topicRouteData.getBrokerDatas()) {
                            String masterAddr = bd.getBrokerAddrs().get(MixAll.MASTER_ID);
                            if (masterAddr != null) {
                                try {
                                    String statsKey = String.format("%s@%s", topic, group);
                                    BrokerStatsData bsd = mqAdminExt.viewBrokerStatsData(masterAddr, BrokerStatsManager.GROUP_GET_NUMS, statsKey);
                                    outTPS += bsd.getStatsMinute().getTps();
                                    outMsgCntToday += StatsAllSubCommand.compute24HourSum(bsd);
                                }
                                catch (Exception e) {
                                    log.warn("Exception caught: mqAdminExt get broker stats data GROUP_GET_NUMS failed");
                                    log.warn("Response [{}] ", e.getMessage());
                                }
                            }
                        }
                    }
                }

                List<String> list;
                try {
                    list = collectService.getTopicMap().get(topic);
                }
                catch (ExecutionException e) {
                    throw Throwables.propagate(e);
                }
                if (null == list) {
                    list = Lists.newArrayList();
                }

                list.add(date.getTime() + "," + new BigDecimal(inTPS).setScale(5, BigDecimal.ROUND_HALF_UP) + "," + inMsgCntToday + "," + new BigDecimal(outTPS).setScale(5, BigDecimal.ROUND_HALF_UP) + "," + outMsgCntToday);
                collectService.getTopicMap().put(topic, list);

                collectService.getTopicSet().add(topic);
            }

            log.debug("Topic Collected Data in memory = {}" + JsonUtil.obj2String(collectService.getTopicMap().asMap()));
        }
        catch (Exception err) {
            throw Throwables.propagate(err);
        }
    }


}
